package com.cpr.core.rabbitmq.client.publishercofirms;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @author Chen Panrui
 * created at 2022/3/16 15:49
 *
 * waitForConfirmsOrDie方法 Broker端在返回nack(Basic.Nack)之后该方法会抛出java.io.IOException
 *
 */
public class ProducerPublisherConfirmsCallBackAck {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        // 指定协议 用户名+密码+host+端口号+虚拟主机
        factory.setUri("amqp://admin:123456@106.13.89.7:5672/%2f");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        // 当前通道标记为 发送方确认机制的通道
        channel.confirmSelect();

        channel.queueDeclare("queue.ppc", true, false, false, null);
        channel.exchangeDeclare("ex.ppc", BuiltinExchangeType.DIRECT, true, false, null);
        channel.queueBind("queue.ppc", "ex.ppc", "key.ppc");

        // 多线程安全的 treeMap，用于存放待确认的消息，类似记录生产者 消费者模型的方式
        ConcurrentNavigableMap<Long, String> confirms = new ConcurrentSkipListMap<>();

        // 处理 broker确认 和 没有被broker确认消息
        channel.addConfirmListener(new ConfirmCallback() {
            // ack的确认
            @Override
            public void handle(long deliveryTag, boolean multiple) throws IOException {
                // deliveryTag 消息编号，multiple 是否有多个消息
                if(multiple){
                    System.out.println("编号<=" + deliveryTag + "的消息都已经被确认了");
                    ConcurrentNavigableMap<Long, String> headMap = confirms.headMap(deliveryTag, true);
                    // 批量移除被确认的消息
                    headMap.clear();
                } else {
                    System.out.println("编号=" + deliveryTag + "的消息已经被确认了");
                    // 移除被确认的消息
                    confirms.remove(deliveryTag);
                }
            }
        }, new ConfirmCallback() {
            // nack确认
            @Override
            public void handle(long deliveryTag, boolean multiple) throws IOException {
                // 将未被确认的消息 放入一个集合中再自个儿处理
                if(multiple){
                    System.out.println("消息编号<=" + deliveryTag + "的消息不被broker确认");
                } else {
                    System.out.println("消息编号=" + deliveryTag + "的消息不被broker确认");
                }
            }
        });

        String message = "-你好";
        for (int i = 0; i < 100; i++) {
            // 获取到 下一条即将发送的消息的id
            long num = channel.getNextPublishSeqNo();
            channel.basicPublish("ex.ppc", "key.ppc", null, (i+message).getBytes(StandardCharsets.UTF_8));
            System.out.println("消息发送成功，编号=" + num + "，等待broker确认");
            confirms.put(num, (i+message));
        }
        System.out.println("消息全部发送完成...");
        System.out.println("---是否存在未被确认消息" + !confirms.isEmpty());
        Thread.sleep(20000);
        System.out.println("---是否存在未被确认消息" + !confirms.isEmpty());


        channel.close();
        connection.close();
    }
}
